package site.anxuan.seckill.service.impl;

import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import site.anxuan.seckill.common.BusinessException;
import site.anxuan.seckill.common.ErrorCode;
import site.anxuan.seckill.common.Tools;
import site.anxuan.seckill.dao.OrderDao;
import site.anxuan.seckill.dao.SerialNumberDao;
import site.anxuan.seckill.entity.Item;
import site.anxuan.seckill.entity.ItemStockLog;
import site.anxuan.seckill.entity.Order;
import site.anxuan.seckill.entity.SerialNumber;
import site.anxuan.seckill.service.ItemService;
import site.anxuan.seckill.service.ItemStockLogService;
import site.anxuan.seckill.service.ItemStockService;
import site.anxuan.seckill.service.OrderService;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.Date;

@Service
@AllArgsConstructor
public class OrderServiceImpl implements OrderService {

    private final Logger logger = LoggerFactory.getLogger(OrderServiceImpl.class);

    private final RedisTemplate<String, Object> redisTemplate;

    private final RocketMQTemplate rocketMQTemplate;

    private final ItemService itemService;

    private final ItemStockService itemStockService;

    private final ItemStockLogService itemStockLogService;

    private final OrderDao orderDao;

    private final SerialNumberDao serialNumberDao;

    @Override
    public void createLocalTransaction(int userId, int itemId, int amount, Integer promotionId) {
        // 记录库存流水
        ItemStockLog itemStockLog = itemStockLogService.insert(itemId, amount);
        logger.info("库存流水[" + itemStockLog.getId() + "]记录成功.");

        // 发送给本地事务监听器的消息，用来执行一致性消息
        JSONObject body = new JSONObject();
        body.put("itemId", itemId);
        body.put("amount", amount);
        body.put("itemStockLogId", itemStockLog.getId());

        // 本地事务参数
        JSONObject arg = new JSONObject();
        arg.put("userId", userId);
        arg.put("itemId", itemId);
        arg.put("amount", amount);
        arg.put("promotionId", promotionId);
        arg.put("itemStockLogId", itemStockLog.getId());

        Message<String> message = MessageBuilder.withPayload(body.toString()).build();

        // 发给扣减库存的消息监听器
        String destination = "seckill:decrease_stock";
        try {
            logger.info("发送扣减库存的消息，如果本地事务完成.");
            TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, arg);
            if (sendResult.getLocalTransactionState() == LocalTransactionState.UNKNOW) {
                throw new BusinessException(ErrorCode.UNDEFINED_ERROR, "创建订单失败！");
            } else if (sendResult.getLocalTransactionState() == LocalTransactionState.ROLLBACK_MESSAGE) {
                throw new BusinessException(ErrorCode.CREATE_ORDER_FAILURE, "创建订单失败！");
            }
        } catch (Exception e) {
            throw new BusinessException(ErrorCode.CREATE_ORDER_FAILURE, "创建订单失败！");
        }
    }

    @Override
    @Transactional
    public RocketMQLocalTransactionState createOrder(int userId, int itemId, int amount,
                                                     Integer promotionId, String itemStockLogId) {
        try {
            // 检查商品
            Item item = itemService.getByIdFromCache(itemId);
            if (item == null) {
                item = itemService.getById(itemId);
                if (item == null) {
                    throw new BusinessException(ErrorCode.CREATE_ORDER_FAILURE, "商品不存在！");
                }
            }

            // 预减库存
            boolean success = itemStockService.decreaseStockInCache(itemId, amount);
            logger.info("商品[" + itemId + "]预减" + amount + "个库存完成.");
            if (!success) {
                throw new BusinessException(ErrorCode.STOCK_NOT_ENOUGH, "库存不足！");
            }

            // 创建订单
            Order order = new Order();
            order.setId(this.generateOrderID());
            order.setUserId(userId);
            order.setItemId(itemId);
            order.setPromotionId(promotionId);
            order.setOrderPrice(promotionId != null ? item.getPromotion().getPromotionPrice() : item.getPrice());
            order.setOrderAmount(amount);
            order.setOrderTotal(order.getOrderPrice().multiply(new BigDecimal(amount)));
            order.setOrderTime(new Timestamp(System.currentTimeMillis()));
            orderDao.insert(order);
            logger.info("生成订单完成 [" + order.getId() + "]");


            // 异步增加销量
            itemService.asyncIncreaseSales(itemId, amount);

            // 修改流水状态
            itemStockLogService.updateStatus(itemStockLogId, ItemStockLog.State.SUCCESS);

            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            logger.error(e.getMessage() + "订单生成失败.");
            itemStockLogService.updateStatus(itemStockLogId, ItemStockLog.State.FAILURE);
            logger.error("更新流水[" + itemStockLogId + "]为失败状态.");
            itemStockService.increaseStockInCache(itemId, amount);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    protected String generateOrderID() {
        StringBuilder sb = new StringBuilder();
        sb.append(Tools.dateFormat(new Date(), "yyyyMMdd"));

        SerialNumber serial = serialNumberDao.getByName("order_serial");
        Integer value = serial.getValue();

        serial.setValue(value + serial.getStep());
        serialNumberDao.updateByName(serial);

        String prefix = "000000000000".substring(value.toString().length());
        sb.append(prefix).append(value);

        return sb.toString();
    }

}
